/* $begin tinymain */
/*
 * tiny.c - A simple, iterative HTTP/1.0 Web server that uses the 
 *     GET method to serve static and dynamic content.
 */


#include <sys/types.h>
#include <sys/socket.h>
#include <netdb.h>
#include <unistd.h>
#include <fcntl.h>
#include <sys/epoll.h>
#include <errno.h>
#include <sys/epoll.h>
#include <string>
#include <syslog.h>
#include "ads_test_mgr.h"
#include "csapp.h"
#include "sbuf.h"
#define NTHREADS  4
#define SBUFSIZE  16
#define MAX_NTHREADS 1024
#define THREAD_DEBUG 0

using namespace std;
#define XML_FILE_NAME_SEND "../xml/GPS_AssistData.xml"

char *g_test_xml_str ="<?xml version=\"1.0\" encoding=\"UTF-8\"?>\
		<!DOCTYPE Eadp SYSTEM \"EADP_Message.dtd\">\
		<Eadp>\
		<Header ver=\"2.0.0\">\
		<SequenceNum>1234</SequenceNum>\
		<ReqSeqNum>1000</ReqSeqNum>\
		</Header>\n\
		<GANSSReport>\
		<GANSSAssistData>12312\
		</GANSSAssistData>\n\
		</GANSSReport>\n\
		</Eadp>\n";


sbuf_t sbuf; /* shared buffer of connected descriptors */
enum BOOL {FALSE,TRUE};
typedef struct{
	pthread_t tid; // thread  tid
	int b_exit;// the tag for exit this thread, true for exit the thread, 0 for continue running
	int b_running; // business in running not on wait;
}thread_status;

sem_t strategy_mutex;       /* Notify manager thread run the strategy */
sem_t ads_test_mutex;       /* ads test mutex , 10 sec time out */
sem_t tids_status_nums_mutex; 	    /*proect tids_exit_tags and current_num_thrs*/
int current_num_thrs = NTHREADS;
thread_status *tids_exit_tags = NULL;
void doit(int fd);
void *doit_thread(void *vargp);
void *manager_thread(void *vargp);
void *ads_test_thread(void *vargp);
void manager_threads(thread_status *tid_status,int buffer_counter, int *nums_thrs);
void debug_command();
void doit_thread_cleanup(void *arg);
int read_write_socket( int fd );
int create_new_conn(int epollfd, int cnn_number, char* host, int port,struct epoll_event& ev);
int accept_sockets(int epollfd, int sfd, struct epoll_event &event);
tr1::shared_ptr<CUserMgr> g_user_mgr;


int make_socket_non_blocking (int sfd)
{
	int flags, s;

	flags = fcntl (sfd, F_GETFL, 0);
	if (flags == -1)
	{
		perror ("fcntl");
		return -1;
	}

	flags |= O_NONBLOCK;
	s = fcntl (sfd, F_SETFL, flags);
	if (s == -1)
	{
		perror ("fcntl");
		return -1;
	}

	return 0;
}

/**
 * 1) find empty slot in g_user_mgr
 * 2) try to create cnn_number of connection on host: port
 * return created number
 */
int main(int argc, char **argv) 
{
	int i,port, epollfd,s;

	char *host;
	int cnn_number = 0;//connection nubmer
	thread_status tids_tags[4] = {{0,0,0}};
	pthread_t tid;

	if (argc != 4) {
		fprintf(stderr, "usage: %s <host> <port> <connection number>\n", argv[0]);
		exit(0);
	}
    host = argv[1];
    port = atoi(argv[2]);
    cnn_number = atoi(argv[3]);
    if(port < 0 || cnn_number < 0 || cnn_number > MAXEVENTS)
    {
		fprintf(stderr, "usage: %s <host> <port> <connection number:65536>\n", argv[0]);
		exit(0);
    }
    openlog(NULL, LOG_CONS | LOG_PID, 0);
    syslog(LOG_INFO,"tiny test mps start\n");
	sbuf_init(&sbuf, SBUFSIZE); //line:conc:pre:initsbuf
	tr1::shared_ptr<CUserMgr> temp(new CUserMgr(host,port, 1000));
	g_user_mgr = temp;
	if(!g_user_mgr)
	{
		printf("new key struct CUserMgr error, exit!\n");
		abort();
	}
	g_user_mgr->create_connection(cnn_number);

	for (i = 0; i < NTHREADS; i++)  /* Create worker threads */ //line:conc:pre:begincreate
		Pthread_create(&tid, NULL, doit_thread, &tids_tags[i]);               //line:conc:pre:endcreate


	struct epoll_event events[MAXEVENTS];
	epollfd = g_user_mgr->get_epollfd();
	int ready;
	while(1)
	{
		ready = epoll_wait(epollfd,events,MAXEVENTS,-1);
		if(ready == -1)
		{
			printf( "Error:epoll_wait fail\n");
			continue;
		}
		for(i = 0 ; i < ready ; i ++)
		{
			if(events[i].data.fd == STDIN_FILENO)
			{
				debug_command();
			}else
			{
				sbuf_insert(&sbuf, events[i].data.fd);
			}
		}
	}
	closelog();
}

void *manager_thread(void *vargp)
{
	tids_exit_tags = (thread_status*) Malloc(sizeof(thread_status)*MAX_NTHREADS);
	memset(tids_exit_tags,0,sizeof(thread_status)*MAX_NTHREADS);
	pthread_t self_tid = pthread_self();
	Pthread_detach(self_tid);
	int vCounter = 0;
	struct timespec ts;
	while(1)
	{
		if (clock_gettime(CLOCK_REALTIME, &ts) == -1)
			printf("clock_gettime");
		ts.tv_sec += 10;
		sem_timedwait(&strategy_mutex, &ts);
		Sleep(1);//selp half second and let cpu go others threads
		if(++vCounter == SBUFSIZE)
		{
			vCounter = 0;//reset the counter
			int item_counter = sbuf_item_counter(&sbuf);// check whether empty or full
			manager_threads(tids_exit_tags, item_counter, &current_num_thrs);
		}
	}
	free(tids_exit_tags);

}

/**
 * manager_threads
 * increase 4 when the buffer becomes full,
 * or decrease the number of threads when the buffer becomes empty step by step
 */
void manager_threads(thread_status *tid_status,int buffer_counter, int *nums_thrs)
{
	int i = 0;
	int l_nums_thread = *nums_thrs;
	int num_threads_changed = 0;
	int num_threads_not_working = 0;
	printf("manager_threads buffer_counter %d,nums_thrs %d, tid_status %p \n",buffer_counter, (*nums_thrs), tid_status);
	P(&tids_status_nums_mutex);
	if( buffer_counter == 0 && *nums_thrs > NTHREADS)
	{// kill threads
		i = 0;
		while(i < MAX_NTHREADS)
		{
			i++;
			if(tid_status[i].tid && !tid_status[i].b_exit)
			{
				if(!tid_status[i].b_running)
				{
					if(0 != pthread_cancel(tid_status[i].tid))
					{
						printf("pthread_cancel error!");
					}
					tid_status[i].b_exit = TRUE;
					//decrease the number of thread by 4 one time
					if(num_threads_changed++ > NTHREADS )
					{
						break;
					}
				}

			}
		}
	}else if( buffer_counter == SBUFSIZE )
	{// increase threads
		i = 0;

		while(i < MAX_NTHREADS)
		{
			i++;
			if( tid_status[i].tid == 0 && sbuf_item_counter(&sbuf) == SBUFSIZE )
			{
				tid_status[i].b_exit = FALSE;
				tid_status[i].b_running = FALSE;
				Pthread_create(&(tid_status[i].tid), NULL, doit_thread, &tid_status[i]);               //line:conc:pre:endcreate
				printf("Create tid %08x \n", tid_status[i].tid);
				//increate the number of thread by NTHREADS one time
				if(num_threads_changed++ > NTHREADS )
				{
					break;
				}
				// let CPU go to others threads after create one thread
				sleep(1);
			}else if(tid_status[i].tid && !tid_status[i].b_running)
			{
				num_threads_not_working++;
			}
		}
		sleep(4);
		*nums_thrs += num_threads_changed;
	}
	V(&tids_status_nums_mutex);
}

void debug_command()
{
	char buf[MAXLINE];
	string debug_str;;
	string second_parameter;
	string first_parameter;
	size_t pos = 0;
	int choise = 0;
	if(!Fgets(buf, MAXLINE,stdin))
		exit(0);
	if(strlen(buf) <= 1)
		goto label_exit;

	debug_str = buf;
	pos = debug_str.find(':');
	if(pos != string::npos)
	{
		first_parameter = debug_str.substr(0,pos);
		second_parameter = debug_str.substr(pos+1);
	}else{
		first_parameter = debug_str;
		second_parameter.clear();
	}

	if(first_parameter.size())
		choise = atoi(first_parameter.c_str());
	else
		choise = -1;
	switch(choise)
	{
	case 0:
		exit(0);
		break;
	case 1:
		printf("send subscribe file to all connection\n");
		if(second_parameter.size() > 0)
			g_user_mgr->send_xml_and_change_state(second_parameter,ADS_USER_SENT_SUB);
		else
			g_user_mgr->send_xml_and_change_state("../xml/Subscribe_Req.xml",ADS_USER_SENT_SUB);
		break;
	case 2:
		printf("send un-subscribe file to all connection\n");
		if(second_parameter.size() > 0)
			g_user_mgr->send_xml_and_change_state(second_parameter,ADS_USER_SENT_UN_SUB);
		else
			g_user_mgr->send_xml_and_change_state("../xml/Unsubscribe.xml",ADS_USER_SENT_UN_SUB);
		break;
	case 3:
		printf("send data request file to all connection\n");
		if(second_parameter.size() > 0)
			g_user_mgr->send_xml_and_change_state(second_parameter,ADS_USER_SENT_DATA_REQ);
		else
			g_user_mgr->send_xml_and_change_state("../xml/DataRequest_GPS.xml",ADS_USER_SENT_DATA_REQ);
		break;
	case 4:
		printf("show_statistics\n");
   		g_user_mgr->show_statistics(!second_parameter.empty());
		break;
	case 5:
		printf("create new connection\n");
   		g_user_mgr->create_connection(atoi(second_parameter.c_str()));
		break;
	default:
		printf("debug_command wrong input, try again\n");
	}
label_exit:
	printf("debug command help:\n"
			"\t0: exit\n"
			"\t1: send subscribe file or default file\te.g:    1:./xml/subscribe.xml\n"
			"\t2: send un-subscribe file or default file\te.g: 2:./xml/un-scribe.xml\n"
			"\t3: send data request file or default file\te.g: 3:./xml/datarequest.xml\n"
			"\t4: show statistics\n"
			"\t5: Create connection \te.g: 5:100, create 100 new connection\n"
			);

}

/**
 * doit_thread_cleanup
 * run doit_thread_cleanup when doit_thread exit
 * purpose:
 *    1)reset tid array
 *    2)decrease current_num_thrs atomic
 */
void doit_thread_cleanup(void *arg)
{
	P(&tids_status_nums_mutex);
	printf("doit_thread_cleanup: 0x%p\n",arg);
	thread_status* tid = (thread_status*)arg;
	tid->tid = 0;
	tid->b_exit = 0;
	tid->b_running = FALSE;
	current_num_thrs--;
	V(&tids_status_nums_mutex);
}
/**
 * tiny web server services thread
 * 1)get description from buffer
 * 2)set thread cancel disable
 * 3)services the request
 * 4)set thread cancel enable
 *
 */
void *doit_thread(void *vargp) 
{ 
	pthread_cleanup_push(doit_thread_cleanup,vargp);
	thread_status* tid = (thread_status*)vargp;
	pthread_t self_tid = pthread_self();
	Pthread_detach(self_tid);

	while (1) {
		tid->b_running = FALSE;
		int connfd = sbuf_remove(&sbuf); /* Remove connfd from buffer */ //line:conc:pre:removeconnfd
		tid->b_running = TRUE;
		pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);
		doit(connfd); 
		pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
		if(sbuf_item_counter(&sbuf) == 0 && tid->b_exit)
		{
			break;
		}
	}
	pthread_cleanup_pop(1);

}

/*
 * doit - handle one socket request/response transaction
 */

void doit(int fd) 
{
	ads_ptr user;
	if((user = g_user_mgr->get_user(fd))!= NULL)
	{
		user->do_ads_action();
	}
}


